package edu.wisc.hevea.hsql;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;


public class MessageDAO {

	private static Log log = LogFactory.getLog(MessageDAO.class);
	
	private static String DB_PATH_PREFIX = "/home/anand/Documents/DistributedSystems/HeveaLibraries/hsql_db/";
	private static String USER_NAME = "root";
	private static String PASSWORD = "scotttiger";

	private Connection dbConnection;
	private String hostName;
	private String dbPath;

	private static MessageDAO _messageDAO;

	public static String MY_MSG_TABLE_NAME = "MY_MSGS";
	public static String REPLICATED_MSG_TABLE_NAME = "REPLICATED_MSGS";
	private static long IDLE_WINDOW_FOR_ME = 1 * 1000;
	private static long IDLE_WINDOW_FOR_REPLICATED = 90 * 1000;
	

	private static String OBJECT_NOT_FOUND_ERR_MSG = "user lacks privilege or object not found";

	private MessageDAO(String hostName) {
		this.hostName = hostName;
		initializeDBConnection(this.hostName);
		try {
			checkAndCreateTable(MY_MSG_TABLE_NAME);
			checkAndCreateTable(REPLICATED_MSG_TABLE_NAME);
		} catch (SQLException e) {
			e.printStackTrace();
		}

	}

	private void initializeDBConnection(String hostName) {
		try {
			Class.forName("org.hsqldb.jdbcDriver");
			dbPath = String.format("%s%s.db", DB_PATH_PREFIX, hostName);
			dbConnection = DriverManager.getConnection(
					getHSQLFileDBConnectionString(dbPath), USER_NAME, PASSWORD);
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		} catch (SQLException e) {
			e.printStackTrace();
		}
	}

	private String getHSQLFileDBConnectionString(String dbPath) {
		return String.format("%s:%s:%s:%s", "jdbc", "hsqldb", "file", dbPath);
	}

	public synchronized static MessageDAO getMessageDAO(String hostName) {
		if (_messageDAO == null) {
			_messageDAO = new MessageDAO(hostName);
		}
		return _messageDAO;
	}

	public void checkAndCreateTable(String tableName) throws SQLException {
		Statement stmt = dbConnection.createStatement();
		try {
			stmt.executeQuery("SELECT * FROM " + tableName + ";");
		} catch (java.sql.SQLSyntaxErrorException ex) {
			if (ex.getMessage().contains(OBJECT_NOT_FOUND_ERR_MSG)) {
				log.info(" Creating table! " + tableName);
				stmt = dbConnection.createStatement();
				stmt.execute("CREATE TEXT TABLE " + tableName + "("
						+ "    MSG_ID   VARCHAR(256) PRIMARY KEY"
						+ "  , MSG_TIME TIMESTAMP DEFAULT CURRENT_TIMESTAMP"
						+ "  , TOPIC    VARCHAR(128)"
						+ "  , CONTENT  VARCHAR(1024)"
						+ "  , STATUS   VARCHAR(32) );");

				stmt.execute("SET TABLE " + tableName + " SOURCE \"" + hostName
						+ "_" + tableName + ".csv;fs=|\";");
			} else {
				throw new SQLException(ex);
			}
		}
	}

	public void updateMessageToConsumed(String tableName, Message m)
			throws SQLException {
		String updateSql = " update " + tableName
				+ " set STATUS = ? where MSG_ID = ?;";
		PreparedStatement updateMsg = dbConnection.prepareStatement(updateSql);
		short columnCount = 1;
		updateMsg.setString(columnCount++, "CONSUMED");
		updateMsg.setString(columnCount++, m.getMsgId());
		updateMsg.executeUpdate();
	}
	
	public void updateReplicatedMessageToConsumed(String msgId)
			throws SQLException {
		String updateSql = " update " + REPLICATED_MSG_TABLE_NAME
				+ " set STATUS = ? where MSG_ID = ?;";
		PreparedStatement updateMsg = dbConnection.prepareStatement(updateSql);
		short columnCount = 1;
		updateMsg.setString(columnCount++, "CONSUMED");
		updateMsg.setString(columnCount++, msgId);
		updateMsg.executeUpdate();
	}
	
	public void insertMessage(String tableName, Message m) throws SQLException {
		String insertSql = " insert into " + tableName
				+ "(MSG_ID,TOPIC,CONTENT,STATUS)" + "values (?,?,?,?); ";
		PreparedStatement insertMsgs = dbConnection.prepareStatement(insertSql);
		short columnCount = 1;
		insertMsgs.setString(columnCount++,
				m.getMsgId());
		insertMsgs.setString(columnCount++, m.getTopic());
		insertMsgs.setString(columnCount++, m.getContent());
		insertMsgs.setString(columnCount++, m.getStatus());
		insertMsgs.executeUpdate();
	}
	
	public void insertMessage(String tableName, String msgId, String topic,
			String content, String status) throws SQLException {
		long currentTime = new Date().getTime();
		String insertSql = " insert into " + tableName
				+ "(MSG_ID,TOPIC,CONTENT,STATUS)" + "values (?,?,?,?); ";
		PreparedStatement insertMsgs = dbConnection.prepareStatement(insertSql);
		short columnCount = 1;
		insertMsgs.setString(columnCount++,
				String.format("%s_%s", hostName, currentTime));
		insertMsgs.setString(columnCount++, topic);
		insertMsgs.setString(columnCount++, content);
		insertMsgs.setString(columnCount++, status);
		insertMsgs.executeUpdate();
	}

	private void convertFromResultSetToMsgList(List<Message> msgList,ResultSet rs) throws SQLException {
		while (rs.next()) {
			Message msg = new Message();
			msg.setMsgId(rs.getString("MSG_ID"));
			msg.setTopic(rs.getString("TOPIC"));
			msg.setContent(rs.getString("CONTENT"));
			msg.setStatus(rs.getString("STATUS"));
			msg.setMsgTime(rs.getTimestamp("MSG_TIME"));
			msgList.add(msg);
		}
	}
	
	public List<Message> getAllPendingMessages(int topN) throws SQLException {

		String selectPendingMsgSql = "SELECT * FROM " + MY_MSG_TABLE_NAME
				+ " where STATUS = ? and MSG_TIME < ? UNION "
				+ " SELECT * FROM " + REPLICATED_MSG_TABLE_NAME
				+ " where STATUS = ? and MSG_TIME < ? order by MSG_TIME;";

		PreparedStatement selectPendingMsgs = dbConnection
				.prepareStatement(selectPendingMsgSql);
		short columnCount = 1;
		selectPendingMsgs.setString(columnCount++, "PENDING");
		Timestamp ts = new Timestamp(new Date().getTime() - IDLE_WINDOW_FOR_ME);
		selectPendingMsgs.setTimestamp(columnCount++, ts);
		selectPendingMsgs.setString(columnCount++, "PENDING");
		ts = new Timestamp(new Date().getTime() - IDLE_WINDOW_FOR_REPLICATED);
		selectPendingMsgs.setTimestamp(columnCount++, ts);

		ResultSet rs = selectPendingMsgs.executeQuery();
		List<Message> pendingMsgList = new ArrayList<Message>();
		convertFromResultSetToMsgList(pendingMsgList, rs);
		return pendingMsgList;
	}
	
	
	public void getAllMessages(String tableName) throws SQLException {
		Statement stmt = dbConnection.createStatement();
		ResultSet rs = stmt.executeQuery("SELECT * FROM " + tableName + ";");
		while (rs.next()) {
			System.out.println("MSG_ID: " + rs.getString("MSG_ID")
					+ "MSG_TIME: " + rs.getTimestamp("MSG_TIME") + "TOPIC: "
					+ rs.getString("TOPIC") + "CONTENT: "
					+ rs.getString("CONTENT") + "STATUS: "
					+ rs.getString("STATUS"));
		}
	}
}
